package com.feidee.bigdata.bloomfilterserv.dao;

import com.feidee.bigdata.bloomfilterserv.constant.Constant;
import com.feidee.fdcommon.util.MD5Util;
import com.feidee.fdhadoop.hive.HiveUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

public class RestoreBfContentThread implements Runnable {

	private Logger logger = LoggerFactory.getLogger(RestoreBfContentThread.class);

	private static final String NOTIFY_FLAG = "消息中心";
	private static final String MSG_FLAG = "短信平台";
	private static final String EMAIL_FLAG = "邮件中心";

	private String restoreNotificationSql;
	private String restoreMsgSql;
	private String restoreEmailSql;

	private String url;
	private String username;
	private String password;

	private IBloomFilterDao bloomFilterDao;

	public RestoreBfContentThread(IBloomFilterDao bloomFilterDao, String restoreNotificationSql, String restoreMsgSql,
	                              String restoreEmailSql, String url, String username, String password) {

		this.restoreNotificationSql = restoreNotificationSql;
		this.restoreMsgSql = restoreMsgSql;
		this.restoreEmailSql = restoreEmailSql;
		this.url = url;
		this.username = username;
		this.password = password;
		this.bloomFilterDao = bloomFilterDao;

	}

	@Override
	public void run() {
		Connection conn = null;
		PreparedStatement stmt = null;
		ResultSet resultSet = null;
		try {
			conn = HiveUtil.getConn("org.apache.hive.jdbc.HiveDriver", this.url, this.username, this.password);

			logger.info("############# 恢复消息布隆结果SQL:" + restoreNotificationSql);
			stmt = conn.prepareStatement(this.restoreNotificationSql);
			resultSet = stmt.executeQuery();
			executeResult(resultSet);
			if (null != stmt) stmt.close();
			if (null != resultSet) resultSet.close();

			logger.info("############# 恢复短信布隆结果SQL:" + restoreMsgSql);
			stmt = conn.prepareStatement(this.restoreMsgSql);
			resultSet = stmt.executeQuery();
			executeResult(resultSet);
			if (null != stmt) stmt.close();
			if (null != resultSet) resultSet.close();

			logger.info("############# 恢复邮件布隆结果SQL:" + restoreEmailSql);
			stmt = conn.prepareStatement(this.restoreEmailSql);
			resultSet = stmt.executeQuery();
			executeResult(resultSet);

		} catch (Exception e) {
			logger.error("恢复布隆过滤服务 Exception,", e);
		} finally {
			try {
				if (null != resultSet) {
					resultSet.close();
				}
				if (null != stmt) {
					stmt.close();
				}
				if (null != conn) {
					conn.close();
				}
			} catch (Exception e1) {
				logger.error("清理数据库连接 Exception", e1);
			}
		}
	}

	/**
	 * 处理结果库
	 *
	 * @param resultSet
	 */
	private void executeResult(ResultSet resultSet) throws SQLException {

		int i = 0; //计数
		String rowKey = "";
		String rowKeyMd5 = "";
		String receiver = "";
		while (resultSet.next()) {

			rowKey = resultSet.getString("rowKey");
			receiver = resultSet.getString("receiver");
			rowKeyMd5 = MD5Util.string2MD5(rowKey);
			int receiverHashCode = MD5Util.string2MD5(receiver).hashCode();
			if (Math.abs(receiverHashCode % Constant.BL_SERVER_NUM) == Constant.BL_HASH_NO) {
				bloomFilterDao.addMetaToBF(rowKeyMd5);

				if (i % 100000 == 0) {
					logger.info("    添加数据:" + receiver + ",rowKey:" + rowKey + ",rowKeyMD5:" + rowKeyMd5 +
							",HashCode:" + receiverHashCode + " 到布隆过滤器中>>>");
					logger.info("添加 " + i + " 条记录到布隆过滤器中...");
				}
				i += 1;
			}

		}
		logger.info("总共添加 " + i + " 条记录到布隆过滤器!!!");

	}
}
